package com.czl.myRabbitMq.mqServer.dataCenter;

import com.czl.myRabbitMq.commom.Consume;
import com.czl.myRabbitMq.commom.ConsumeEnv;
import com.czl.myRabbitMq.commom.exception.mqException;
import com.czl.myRabbitMq.mqServer.VirtualHost;
import com.czl.myRabbitMq.mqServer.core.MSGQueue;
import com.czl.myRabbitMq.mqServer.core.Message;
import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;

/**
 * 这个类主要用来进行消费者的管理
 */
@Slf4j
public class ConsumeManager {
    //需要把VirtualHost这个类加载进来 方便调用内存和硬盘上的数据
    private final VirtualHost virtualHost;

    //线程池来处理 消息的回调函数的逻辑
    private ExecutorService pools = Executors.newFixedThreadPool(5);
    //这个阻塞队列表示 要是当前队列中有消息新增 就把队列名放到当前队列中来
    //以便扫描线程的查找
    private BlockingQueue<String> tokenQueue = new LinkedBlockingQueue<>();
    //扫描队列 不停对BlockingQueue进行扫描 判断队列中是否新增元素
    private Thread scannerThread = null;

    public ConsumeManager(VirtualHost virtualHost){
        this.virtualHost = virtualHost;

        //初始化扫描线程
        scannerThread = new Thread(()->{
            //扫描线程是一直运行的 除非程序结束 不会中断
            while (true){
                try {
                    //先获取到对应的队列
                    String queueName = tokenQueue.take();
                    MSGQueue queue = virtualHost.getMemoryDataManager().selectQueue(queueName);
                    if(queue == null){
                        throw new mqException("扫描线程中，对应交换机id不存在" + queueName);
                    }
                    //从这个队列中消费一个消息
                    synchronized (queue) {
                        consumeMessage(queue);
                    }
                }catch (Exception e){
                    log.info("扫描线程出现异常");
                    e.printStackTrace();
                }
            }
        });
        //将扫描线程设为后台线程 就是该线程不会影响程序的结束和执行
        //前台线程没结束的时候程序无法结束
        scannerThread.setDaemon(true);
        scannerThread.start();
    }

    //往阻塞队列中插入一个元素
    public void notifyConsume(String token) throws InterruptedException {
        tokenQueue.put(token);
    }

    //往队列添加一个消费者
    public void addConsume(String consumeTag, String virtualQueueName, boolean autoACK, Consume consume) throws mqException {
        //先在内存中查找是否存在对应的队列
        MSGQueue queue = virtualHost.getMemoryDataManager().selectQueue(virtualQueueName);

        if(queue == null){
            throw new mqException("添加消费者的队列不存在" + virtualQueueName);
        }

        //创建出消费者对象
        synchronized (this) {
            ConsumeEnv consumeEnv = new ConsumeEnv(consumeTag, queue, autoACK, consume);
            queue.addConsumeEnv(consumeEnv);
            //要是添加消费者的时候当前队列已经存在消息 则让当前消费者 快速进行消息的消费
            int size = virtualHost.getMemoryDataManager().getMessageNum(virtualQueueName);
            for(int i = 0; i < size; i++){
                //这个方法调用一次就消费一条消息
                consumeMessage(queue);
            }
        }
    }

    private void consumeMessage(MSGQueue queue) {
        //先找到当前队列是那个消费者来进行消费
        ConsumeEnv luckyConsumer = queue.chooseConsume();
        if(luckyConsumer == null){
            //表示当前队列中没有消费者 消息就会屯着
            log.info("当前队列中没有消费者" + queue.getName());
            return;
        }
        //获取到需要消费的消息
        Message message = virtualHost.getMemoryDataManager().pollMessage(queue.getName());
        if (message == null){
            log.info("当前队列不存在消费者" + queue.getName());
            return;
        }
        //调用线程池来对 每条消息进行消费 具体的消费逻辑由每个消费者的回调函数来决定
        pools.submit(()->{
            //因为我们前面已经取出消息了 所以要是处理消息的时候抛出异常 消息就会丢失
            //这里我们使用前面的waitACK队列进行消息的一个存储
            //要是 消费者的autoACK = true 则我们自动将该消息删除
            //消费者的autoACK = false 需要消费者在回调函数中手动调用basicACK方法后才进行删除
            try {
                virtualHost.getMemoryDataManager().addMessageWaitACK(queue, message);

                //这里是让线程池对消费者提供的方法进行回调
                luckyConsumer.getConsume().delivery(luckyConsumer.getConsumeTag(),
                        message.getProperties(), message.getMessage());

                //消费者回调函数执行成功且没抛出异常 且是自动删除的逻辑
                if(luckyConsumer.isAutoACK()){
                    //1表示可持久化存储 需要删除硬盘上的数据
                    if(message.getDurable() == 1){
                        virtualHost.getDiskDataManager().deleteMessage(message, queue);
                    }
                    //最后删除等待队列和消息中心的内容 因为上述已从队列中取走消息 所以不需要进行删除
                    virtualHost.getMemoryDataManager().deleteMessageWaitACK(queue.getName(), message.getMessageId());

                    virtualHost.getMemoryDataManager().deleteMessage(message.getMessageId());
                    log.info("消息消费成功" + message.getMessageId());

                }

            }catch (Exception e){
                log.info("消息消息过程中出现异常");
                e.printStackTrace();
            }
        });
    }




}
